package com.mapbased.sfw.common;

import java.io.IOException;
import java.util.concurrent.ExecutorService;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;

import com.mapbased.sfw.util.concurrent.JMXConfigurableThreadPoolExecutor;
import com.mapbased.sfw.util.logging.ESLogger;
import com.mapbased.sfw.util.logging.Loggers;

public class HbaseClient {

	public final static int DEFAULT_SCAN_CACHE = 4000;
	public final static int DEFAULT_SCAN_MULTY_FILTER_BATCH = 200000;

	public static interface DataChanger {
		public void changeData(HTable table);

		public void errorCaught(Throwable t);
	}

	public static interface DataGetter {
		public Object getData(HTable table) throws IOException;

	}

	static private ESLogger log = Loggers.getLogger(HbaseClient.class);
	private HTablePool pool;
	final private ExecutorService fsmPool = new JMXConfigurableThreadPoolExecutor(
			1, 10, "hbaseFSM");
	// Executors.newFixedThreadPool(10);

	public static final HbaseClient INSTANCE = new HbaseClient();

	static boolean hbaseWAL = Config.get()
			.getBoolean("hbase.wal.enabled", true);

	private Configuration hbaseConfiguration;
	HBaseAdmin admin;

	static public Configuration getHBaseConfiguration(Config c) {

		String ourzk = c.get("zk.quorum", "");
		String[] servers = ourzk.split(",");
		StringBuilder bs = new StringBuilder();
		for (String ss : servers) {
			String s[] = ss.split(":");
			if (s[0] != null)
				bs.append(s[0]).append(",");

		}
		bs.setLength(bs.length() - 1);
		// System.out.print("hbase.zookeeper.quorum:" + bs.toString());

		Configuration hbc = new Configuration();
		hbc.set("hbase.zookeeper.property.clientPort", "2181");
		hbc.set("hbase.zookeeper.quorum", bs.toString());
		return hbc;
	}

	public HbaseClient() {

		this.reCreate();
	}

	public void reCreate() {
		Config c = Config.get();
		this.hbaseConfiguration = getHBaseConfiguration(c);

		pool = new HTablePool(hbaseConfiguration, c.getInt(
				"hbase.pool.table.max", 1000));
		try {
			admin = new HBaseAdmin(this.hbaseConfiguration);
		} catch (Exception e) {
			log.error("Error create HBaseAdmin", e);
			System.exit(1);
		}
		log.debug("hbase wal:", hbaseWAL);
	}

	public HTable createHTable(String name) throws IOException {
		return new HTable(this.hbaseConfiguration, name);
	}

	public HTable getHTable(String name) throws IOException {
		return (HTable) pool.getTable(name);
	}

	public void releaseHTable(HTable table) {
		if (table != null)
			pool.putTable(table);
	}

	public Object getData(final String tablename, final DataGetter dc)
			throws IOException {

		HTable table = null;
		try {
			table = (HTable) pool.getTable(tablename);
			return dc.getData(table);
		} finally {
			if (table != null)
				pool.putTable(table);
		}
	}

	public void changeData(final String tablename, final DataChanger dc) {
		fsmPool.submit(new Runnable() {
			public void run() {
				HTable table = null;
				try {
					table = (HTable) pool.getTable(tablename);
					table.setAutoFlush(true);

					dc.changeData(table);
				} catch (Throwable t) {
					dc.errorCaught(t);
				} finally {
					if (table != null)
						pool.putTable(table);
				}

			}
		});
	}

	public void dropTable(String tableName) throws MasterNotRunningException,
			IOException {

		if (admin.isTableEnabled(tableName)) {
			admin.disableTable(tableName);
		}
		admin.deleteTable(tableName);
	}

	// public boolean isTableExist(String s)
	// {
	//
	// }

	public void putColumn(String tablename, String row, String family,
			String qualifier, byte[] value) throws IOException {
		Put put = new Put(row.getBytes());
		put.add(family.getBytes(), qualifier.getBytes(), value);
		HTable table = (HTable) pool.getTable(tablename);
		try {
			table.put(put);
			table.flushCommits();
		} finally {
			pool.putTable(table);
		}

	}

	/**
	 * delete column
	 * 
	 * @param tablename
	 * @param row
	 * @param column
	 * @throws IOException
	 */
	public void delColumn(String tablename, String row, String family,
			String qualifier) throws IOException {
		HTable table = (HTable) pool.getTable(tablename);
		Delete del = new Delete(row.getBytes());
		del.deleteColumn(family.getBytes(), qualifier.getBytes());

		try {
			table.delete(del);
			table.flushCommits();
		} finally {
			pool.putTable(table);
		}

	}

	public static void main(String[] args) throws IOException {
		HbaseClient.INSTANCE.getData("z9", new DataGetter()

		{
			public Object getData(HTable table) throws IOException {
				return table.getStartKeys();
			}
		});
		// HbaseClient.INSTANCE.fsmPool.shutdown();;

	}

	static class StringClass {
		public String str;
	}

	public Scan getScan(String tableName, String familyName, long timeStamp,
			Filter filter, String startRow, int cacheSize) throws IOException {
		Scan s = new Scan();
		s.setCacheBlocks(true);
		s.setMaxVersions(1);
		if (familyName != null && !familyName.isEmpty()) {
			s.addFamily(Bytes.toBytes(familyName));
		}
		if (timeStamp > 0) {
			s.setTimeRange(0, timeStamp);
		}
		if (filter != null) {
			s.setFilter(filter);
		}
		if (startRow != null) {
			s.setStartRow(Bytes.toBytes(startRow));
		}
		s.setCaching(cacheSize);
		return s;
	}

	public static interface ResultProcessor<T> {
		T process(Result r) throws Exception;
	}

	private <T> void scan(HTable table, Scan scan, ResultProcessor<T> processor)
			throws Exception {
		ResultScanner scanner = null;
		try {
			scanner = table.getScanner(scan);
			for (Result r : scanner) {
				processor.process(r);
			}
		} finally {
			if (scanner != null) {
				scanner.close();
			}
		}
	}

	public <T> void scan(final String tableName, final Scan scan,
			final ResultProcessor<T> processor) throws Exception {

		getData(tableName, new HbaseClient.DataGetter() {

			@Override
			public Object getData(HTable table) throws IOException {
				try {
					scan(table, scan, processor);
				} catch (Exception e) {
					throw new IOException(e);
				}
				return null;
			}
		});
	}

	public <T> void scan(String tableName, String familyName, long timeStamp,
			Filter filter, String startRow, int cacheSize,
			ResultProcessor<T> processor) throws Exception {
		scan(tableName, getScan(tableName, familyName, timeStamp, filter,
				startRow, cacheSize), processor);
	}

	public <T> void scan(String tableName, final String familyName,
			final ResultProcessor<T> processor) throws Exception {
		scan(tableName, familyName, 0, null, null, DEFAULT_SCAN_CACHE,
				processor);
	}

	public <T> void scan(String tableName, final ResultProcessor<T> processor)
			throws Exception {
		scan(tableName, "", processor);
	}

}
